package org.yuanzheng.transformation

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment

/**
 * @author yuanzheng
 * @date 2020/6/16-9:31
 */
object TestConnect {
  def main(args: Array[String]): Unit = {
    val streamEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    streamEnv.setParallelism(1)
    import org.apache.flink.streaming.api.scala._
    // 不同类型的数据集
    val dataStream1: DataStream[(String, Int)] = streamEnv.fromElements(("a", 3), ("b", 4), ("c", 2), ("c", 5), ("a", 5))
    val dataStream2: DataStream[Int] = streamEnv.fromElements(1, 2, 4, 5, 6)
    // 连接两个dataStream数据集
    val dataStream: ConnectedStreams[(String, Int), Int] = dataStream1.connect(dataStream2)
    //coMap处理需要传入两个function来处理合并的流
    val result: DataStream[(String, Int)] = dataStream.map(t1 => {
      (t1._1, t1._2)
    }, t2 => {
      (t2 + "", 0)
    }).keyBy(0).sum(1)
    //
    result.print()
    //
    streamEnv.execute()
  }
}
